package com.stan.common;

import com.alibaba.fastjson.JSONObject;
import com.stan.core.conf.ConfManager;
import com.stan.core.contants.Constants;
import com.stan.simulate.SimulateDataSource;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;

/**
 * Spark工具类
 * @author Administrator
 *
 */
public class SparkUtils {

	/**
	 * 是否使用本地模式运行
	 * @return
	 */
	public static boolean isLocal() {
		boolean local = false;
		String sparkMaster = ConfManager.getProperty(Constants.CONF_SPARK_MASTER);
		if(StringUtils.isNotEmpty(sparkMaster) && sparkMaster.contains("local")){
			local = true;
		}
		return local;
	}

	
	/**
	 * 根据当前是否本地测试的配置
	 * 决定，如何设置SparkConf的master
	 */
	public static void setMaster(SparkConf conf) {
		boolean local = ConfManager.getBoolean(Constants.CONF_SPARK_LOCAL);
		if(local) {
			conf.setMaster("local");
		}  
	}
	
	/**
	 * 获取SQLContext
	 * 如果spark.local设置为true，那么就创建SQLContext；否则，创建HiveContext
	 * @param sc
	 * @return
	 */
	public static SQLContext getSQLContext(SparkContext sc) {
		SQLContext sqlContext = null;
		if(isLocal()){
			sqlContext = new SQLContext(sc);
		}else{
			sqlContext = new HiveContext(sc);
		}
		return sqlContext;
	}
	
	/**
	 * 生成模拟数据
	 * 如果spark.local配置设置为true，则生成模拟数据；否则不生成
	 * @param sc
	 * @param sqlContext
	 */
	public static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
		boolean local = isLocal();
		if(local) {
			SimulateDataSource.mock(sc, sqlContext);
		}
	}

	/**
	 * 获得指定时间范围的RDD数据(通过SQL查询获得)
	 * @param sqlContext
	 * @return
	 */
	public static JavaRDD<Row> queryUserActionDataToRDDWithDateRange(SQLContext sqlContext, JSONObject taskParams){
		String startDate = taskParams.getString(Constants.PARAM_FIELD_START_DATE);
		String endDate = taskParams.getString(Constants.PARAM_FIELD_END_DATE);
		// TODO: 测试,不指定时间范围
		String sql = "select * from " + ConfManager.getProperty(Constants.TABLE_NAME_USER_ACTION);
//		+
//				" where date >= \'"+startDate+"\' and " +
//				"date <= \'"+endDate +"\'";
		System.out.println(sql);
		DataFrame dataFrame =  sqlContext.sql(sql);
		JavaRDD<Row> rowsRDD = dataFrame.javaRDD();
		return rowsRDD;
	}


}
